/**
 * Copyright © 2011 Creativor Studio.
 * About license information, please see LICENSE.txt.
 */
package org.creativor.rayson.transport.client;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.logging.Level;
import org.creativor.rayson.api.TransferArgument;
import org.creativor.rayson.api.TransferSocket;
import org.creativor.rayson.client.ClientCall;
import org.creativor.rayson.exception.IllegalServiceException;
import org.creativor.rayson.exception.NetWorkException;
import org.creativor.rayson.exception.ServiceNotFoundException;
import org.creativor.rayson.exception.UnsupportedVersionException;
import org.creativor.rayson.transport.common.Packet;
import org.creativor.rayson.util.Log;

// TODO: Auto-generated Javadoc
/**
 * A connector build a bridge between RPC layer and transport layer.
 * 
 * @author Nick Zhang
 */
public class RpcConnector
{
	private static class CallWrapper
	{
		private ClientCall call;
		private long connectionId;
		
		CallWrapper(ClientCall call, long connectionId)
		{
			this.call = call;
			this.connectionId = connectionId;
		}
		
		public ClientCall getCall()
		{
			return call;
		}
		
		public long getConnectionId()
		{
			return connectionId;
		}
	}
	
	private static class ConnectionCalls
	{
		private HashSet<Long> calls;
		private long connectionId;
		
		ConnectionCalls(long connectionId)
		{
			this.connectionId = connectionId;
			this.calls = new HashSet<Long>();
		}
		
		public void addCall(long callId)
		{
			this.calls.add(callId);
		}
		
		/**
		 * @return the connectionId
		 */
		public long getConnectionId()
		{
			return connectionId;
		}
		
		public Iterator<Long> iterator()
		{
			return this.calls.iterator();
		}
		
		public void removeCall(long callId)
		{
			this.calls.remove(callId);
		}
	}
	
	/**
	 * Presents server with the same server key.
	 * 
	 * @author Nick Zhang
	 */
	private static class ServerAgent
	{
		private ServerKey key;
		
		/**
		 * The last one usable {@link RpcConnection}.
		 */
		private RpcConnection lastUableConnection;
		
		ServerAgent(ServerKey key)
		{
			this.key = key;
		}
		
		/**
		 * @return the key
		 */
		public ServerKey getKey()
		{
			return key;
		}
	}
	
	/**
	 * Server key, identified with the server socket address hash code.
	 * 
	 * @author Nick Zhang
	 */
	private static class ServerKey
	{
		private SocketAddress serverAddress;
		
		ServerKey(SocketAddress serverAddress)
		{
			this.serverAddress = serverAddress;
		}
		
		@Override
		public boolean equals(Object obj)
		{
			if (obj == null) return false;
			if (!(obj instanceof ServerKey)) return false;
			return this.serverAddress.equals(((ServerKey) obj).serverAddress);
		}
		
		@Override
		public int hashCode()
		{
			return serverAddress.hashCode();
		}
	}
	
	/**
	 * Map of <Call id, RPC call wrapper>.
	 */
	private HashMap<Long, CallWrapper> calls;
	private TransportClient client;
	/**
	 * Map of <Connection id , RPC Calls>.
	 */
	private HashMap<Long, ConnectionCalls> connections;
	/**
	 * Map of <Server key , Server Agent>.
	 */
	private HashMap<ServerKey, ServerAgent> servers;
	
	/**
	 * Instantiates a new rpc connector.
	 * 
	 * @param client
	 *            the client
	 */
	RpcConnector(TransportClient client)
	{
		this.client = client;
		calls = new HashMap<Long, CallWrapper>();
		connections = new HashMap<Long, RpcConnector.ConnectionCalls>();
		servers = new HashMap<ServerKey, RpcConnector.ServerAgent>();
	}
	
	private synchronized void addCall(ClientCall call, long connectionId)
	{
		this.calls.put(call.getId(), new CallWrapper(call, connectionId));
		ConnectionCalls connectionCalls = connections.get(connectionId);
		if (connectionCalls == null)
		{
			connectionCalls = new ConnectionCalls(connectionId);
			connections.put(connectionId, connectionCalls);
		}
		connectionCalls.addCall(call.getId());
	}
	
	/**
	 * @param serverAddress
	 * @return
	 */
	private ServerAgent createOrGetServerAgent(SocketAddress serverAddress)
	{
		ServerKey serverKey = new ServerKey(serverAddress);
		synchronized (servers)
		{
			ServerAgent serverAgent = servers.get(serverKey);
			if (serverAgent == null)
			{
				serverAgent = new ServerAgent(serverKey);
				servers.put(serverKey, serverAgent);
			}
			
			return serverAgent;
		}
	}
	
	/**
	 * Notify connection closed.
	 * 
	 * @param connection
	 *            the connection
	 */
	protected synchronized void notifyConnectionClosed(RpcConnection connection)
	{
		ConnectionCalls connectionCalls = connections.remove(connection.getId());
		if (connectionCalls == null)
		{
			Log.getLogger().log(Level.SEVERE,
					"Can not find connection calls in connection :" + connection.getId());
			return;
		}
		
		// remove from call list.
		for (Iterator<Long> iterator = connectionCalls.iterator(); iterator.hasNext();)
		{
			Long callId = iterator.next();
			CallWrapper callWrapper = this.calls.remove(callId);
			if (callWrapper == null) continue;
			callWrapper.call.notifyConnectionClosed();
		}
	}
	
	/**
	 * Open transfer socket.
	 * 
	 * @param serverAddress
	 *            the server address
	 * @param argument
	 *            the argument
	 * @return the transfer socket
	 * @throws ConnectException
	 *             the connect exception
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 * @throws ServiceNotFoundException
	 *             the service not found exception
	 * @throws IllegalServiceException
	 *             the illegal service exception
	 * @throws UnsupportedVersionException
	 *             the unsupported version exception
	 */
	public TransferSocket
			openTransferSocket(SocketAddress serverAddress, TransferArgument argument)
					throws ConnectException, IOException, ServiceNotFoundException,
					IllegalServiceException, UnsupportedVersionException
	{
		return client.createTransferSocket(serverAddress, argument);
	}
	
	/**
	 * Ping.
	 * 
	 * @param serverAddress
	 *            the server address
	 * @throws NetWorkException
	 *             the net work exception
	 */
	public void ping(SocketAddress serverAddress) throws NetWorkException
	{
		client.ping(serverAddress);
	}
	
	private synchronized ClientCall removeCall(long callId)
	{
		CallWrapper callWrapper = calls.remove(callId);
		if (callWrapper == null)
		{
			return null;
		}
		// remove from connection calls
		ConnectionCalls connectionCalls = connections.get(callWrapper.connectionId);
		if (connectionCalls == null)
		{
			Log.getLogger().log(Level.SEVERE,
					"Can not find call in connection :" + callWrapper.connectionId);
			return null;
		}
		connectionCalls.removeCall(callId);
		return callWrapper.call;
	}
	
	/**
	 * Response one call.
	 * 
	 * @throws InterruptedException
	 *             the interrupted exception
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	public void responseOneCall() throws InterruptedException, IOException
	{
		Packet receivedPacket = this.client.getPacketManager().takeReceived();
		DataInputStream inputStream =
				new DataInputStream(new ByteArrayInputStream(receivedPacket.getData()));
		long callId = -1;
		callId = inputStream.readLong();
		ClientCall call = removeCall(callId);
		if (call == null)
		{
			return;
		}
		call.readResult(inputStream);
	}
	
	/**
	 * Submit call.
	 * 
	 * @param serverAddress
	 *            the server address
	 * @param call
	 *            the call
	 * @throws ConnectException
	 *             the connect exception
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 * @throws InterruptedException
	 *             the interrupted exception
	 */
	public void submitCall(SocketAddress serverAddress, ClientCall call) throws ConnectException,
			IOException, InterruptedException
	{
		if (serverAddress == null) throw new IllegalArgumentException("Server address is null");
		ServerAgent serverAgent = createOrGetServerAgent(serverAddress);
		RpcConnection connection;
		synchronized (serverAgent)
		{
			// We use only last usable connection of server agent. In the
			// future, we should improve the performance.
			connection = serverAgent.lastUableConnection;
			if (connection == null || !connection.isReusable())
			{
				connection = client.createRpcConnection(serverAddress);
				serverAgent.lastUableConnection = connection;
			}
		}
		
		addCall(call, connection.getId());
		connection.addSendPacket(call.getRequestPacket());
	}
}
